package com.orange.zkcase2;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * Description: zookeeper分布式锁案例
 */
public class DistributedLock {
    //注意：connectString逗号左右不能有空格，否则连接不上
    private final String connectString = "192.168.147.128:2181,192.168.147.129:2181,192.168.147.130:2181";
    //tickTime为2000，initLimit为10
    //LF初次连接时的超时时间应起码大于延迟时间tickTime*initLimit的值否则会因为超时而连接失败。
    private final int sessionTimeout = 200000;
    private final ZooKeeper zkClient;

    //增加代码健壮性
    //zookeeper连接
    private CountDownLatch connectLatch = new CountDownLatch(1);
    //zookeeper等待
    private CountDownLatch waitLatch = new CountDownLatch(1);

    //当前client等待的子节点的路径
    private String waitPath;
    //当前client创建的子节点
    private String currentNode;

    /**
     * 和zk创建连接，并创建根节点
     */
    public DistributedLock() throws IOException, InterruptedException, KeeperException {
        //1.获取连接 建立服务端与客户端连接
        zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                System.out.println("-----process-------");
                // connectLatch 如果连接上zk  可以释放
                // 连接建立时, 打开latch, 唤醒wait在该latch上的线程
                if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
                    connectLatch.countDown();
                }

                // 发生了waitPath的删除事件 需要释放
                if (watchedEvent.getType() == Event.EventType.NodeDeleted && watchedEvent.getPath().equals(waitPath)) {
                    waitLatch.countDown();
                }
            }
        });

        //等待zookeeper正常连接后，代码才往下继续执行
        connectLatch.await();

        //2.判断根节点 /locks 是否存在
        Stat stat = zkClient.exists("/locks", false);
        //如果根节点不存在，则创建根节点，根节点类型为永久节点
        if (stat == null) {
            System.out.println("根节点不存在");
            // 创建根节点,根节点必须是永久节点
            zkClient.create("/locks", "locks".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }

    //对zk 加锁
    public void zkLock() {
        try {
            //创建对应的临时带序号临时节点，返回值为创建的节点路径
            currentNode = zkClient.create("/locks/" + "seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            System.out.println(Thread.currentThread().getName() + "当前节点为：" + currentNode);

            //注意, 没有必要监听"/locks"的子节点的变化情况
            //判断创建的节点是否是最小的序号节点，如果是获取到锁；如果不是，监听它序号前一个节点
            List<String> children = zkClient.getChildren("/locks", false);
            //如果children只要一个子节点，那就直接获取锁； 如果有多个节点，需要判断，谁最小
            if (children.size() == 1) {
                System.out.println(Thread.currentThread().getName() + "对zk 加锁, 当前节点：" + currentNode);
                return;
            } else {
                ///对根节点下的所有临时顺序节点进行从小到大排序，有序递增
                Collections.sort(children);
                //获取当前节点名称 seq-00000000
                String thisNode = currentNode.substring("/locks/".length());
                System.out.println(Thread.currentThread().getName() + "当前节点名称为：" + thisNode);
                // 通过seq-00000000 获取该节点在children集合的位置
                int index = children.indexOf(thisNode);
                System.out.println(Thread.currentThread().getName() + "当前节点在集合的位置为：" + index);

                //判断
                if (index == -1) {
                    System.out.println(Thread.currentThread().getName() + "数据异常");
                } else if (index == 0) {
                    //只有一个节点，就可以获取锁了
                    System.out.println(Thread.currentThread().getName() + "对zk 加锁, 当前节点：" + currentNode);
                    return;
                } else {
                    //获得排名比 currentNode 前 1 位的节点
                    waitPath = "/locks/" + children.get(index - 1);
                    System.out.println(Thread.currentThread().getName() + "前一个节点为：" + waitPath);
                    //在 waitPath 上注册监听器, 当 waitPath 被删除时, zookeeper会回调监听器的 process方法
                    //需要监听 它前一个节点变化
                    zkClient.getData(waitPath, true, null);

                    //入等待锁状态,等待监听
                    waitLatch.await();

                    return;
                }
            }
        } catch (KeeperException e) {
            throw new RuntimeException(e);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    //对zk 解锁
    public void unZkLock() {
        try {
            System.out.println(Thread.currentThread().getName() + "解锁，删除当前节点：" + currentNode);
            //删除节点
            zkClient.delete(currentNode, -1);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (KeeperException e) {
            throw new RuntimeException(e);
        }
    }
}